package mylab.starters.mq.impl.kafka;


import mylab.starters.mq.api.MessageListener;
import mylab.starters.mq.api.Properties;
import mylab.starters.mq.impl.AbstractConsumer;
import mylab.starters.mq.impl.kafka.loop.PullLoop;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;

public class KafkaConsumer extends AbstractConsumer {

    protected org.apache.kafka.clients.consumer.KafkaConsumer<String, String> consumer;
    protected PullLoop loop;

    public KafkaConsumer(Properties properties, String groupId, String clientId) {
        super(properties, groupId, clientId);
    }

    @Override
    protected void onSubscribe(String topic, MessageListener messageListener, String... tags) throws Exception {
        initConsumer();
        consumer.subscribe(Collections.singleton(topic));
        loop = new PullLoop(consumer, messageListener, tags);
        loop.start();

        consumer.subscribe(Collections.singleton(topic));
    }

    @Override
    protected void onUnsubscribe() throws Exception {
        loop.stop();
        loop = null;
    }

    private void initConsumer() {
        java.util.Properties properties = new java.util.Properties();
        //序列化类型配置
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //服务端地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.properties.getBrokers());
        //消费组ID
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        //客户端ID
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
        //是否自动确认offset, 强制由业务层来确认
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);


        consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(properties);
    }
}
